package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo02WordCountSubmit {
  /**
   * 提交任务到集群：
   * 1、setMaster不能再指定为local了，一般会注释
   * 2、注意输入输出路径得换成HDFS的路径，并准备好数据
   * 3、打包上传
   * 4、使用命令进行提交
   * spark-submit --class com.shujia.core.Demo02WordCountSubmit --master yarn-client spark-1.0-SNAPSHOT.jar
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo02WordCountSubmit")
    conf.setMaster("local") // Spark 运行的方式

    // 构建Spark环境：SparkContext
    val sc: SparkContext = new SparkContext(conf)

    // 读取数据
    /**
     * RDD：弹性分布式数据集
     * RDD是不存储数据的，因为在大数据的环境下，数据量比较大，不适合复制
     * RDD实际上是一种抽象的编程模型
     * RDD具有5大特性：
     * 1、RDD是由一系列分区组成的，第一个RDD的分区数，一般情况下由切片的数量决定
     * 特殊情况：当只有一个切片，但最小分区数>1，假设是2，那么分区的数量由最小分区数决定
     * 2、Task是作用在每个分区上的，每个分区都需要一个Task进行处理
     * 3、RDD之间是有一系列依赖关系的
     * 4、分区类算子需要作用在KV格式的RDD上
     * 5、Spark会给每个Task提供最佳的计算位置，移动计算，不移动数据
     */
    val lineRDD: RDD[String] = sc.textFile("data/words/")
    println(lineRDD.getNumPartitions)

    val wordsRDD: RDD[String] = lineRDD
      .flatMap(_.split(","))

    val groupRDD: RDD[(String, Iterable[String])] = wordsRDD
      .groupBy(word => word)


    val wordCntRDD: RDD[String] = groupRDD
      .map(kv => {
        val word: String = kv._1
        val cnt: Int = kv._2.size
        s"$word, $cnt"
      })

    wordCntRDD
      //      .foreach(println) // 在将任务提交到集群运行，一般不会直接打印数据，因为数据量比较大
      // 通常数据处理完之后会保存到某个目录当中
      .saveAsTextFile("data/words/")

    //    while (true) {
    //
    //    }

  }

}
